import TimeWindowTest.TempData
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

object CountWindowTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream1 = env.readTextFile("D:\\codes\\IDEA_prjects\\FlinkDemo\\data\\sensor.txt")
    val stream2 = stream1.map(
      data => {
        val strings = data.split(",")
        TempData(strings(0), strings(1).toLong, strings(2).toDouble)
      }
    )

    //TODO 滚动窗口
    stream2
      .map(r => (r.id, r.temp))
      .keyBy(_._1)
      .countWindow(5)
      .reduce((x, y) => (x._1, x._2.min(y._2)))

    //TODO 滑动窗口
    stream2
      .map(r => (r.id, r.temp))
      .keyBy(_._1)
      //每当某一个 key 的个数达到 10 的时候,触发计算，计算最近该 key 最近 2 个元素的内
      .countWindow(10, 2)
      .sum(1)

    env.execute()
  }
}
